package com.yxsk.relay.job.component.common.net.handler;

import com.yxsk.relay.job.component.common.exception.net.RelayJobNetException;
import com.yxsk.relay.job.component.common.net.handler.convert.NetMessageConvert;
import com.yxsk.relay.job.component.common.net.handler.exception.GlobalNetServerExceptionHandler;
import com.yxsk.relay.job.component.common.net.handler.interceptor.RpcHandleInterceptor;
import com.yxsk.relay.job.component.common.net.handler.message.MessageServerHandler;
import com.yxsk.relay.job.component.common.net.message.NetRequestMessage;
import com.yxsk.relay.job.component.common.net.message.NetResponseMessage;
import lombok.Builder;
import lombok.extern.slf4j.Slf4j;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.DynamicChannelBuffer;
import org.jboss.netty.channel.*;
import org.jboss.netty.handler.codec.http.*;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.*;

/**
 * @Description
 * @Author 11376
 * @CreateTime 2019/9/6 15:41
 */
@Slf4j
@Builder
public class HttpServerHandler extends SimpleChannelUpstreamHandler {

    // relay-job RPC执行处理器
    private MessageServerHandler messageServerHandler;

    private NetMessageConvert messageConvert;

    private List<RpcHandleInterceptor> interceptors;

    private GlobalNetServerExceptionHandler exceptionHandler = new GlobalNetServerExceptionHandler.DefaultGlobalExceptionHandler();

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {

        try {
            // build request message
            NetRequestMessage requestMessage = this.buildRequestMessage(e);

            // before intercept
            if (this.interceptors != null && !interceptors.isEmpty()) {
                Iterator<RpcHandleInterceptor> iterator = this.interceptors.iterator();
                while (iterator.hasNext()) {
                    iterator.next().before(requestMessage);
                }
            }

            // handle
            byte[] message = this.messageServerHandler.handle(requestMessage);

            // build response message
            NetResponseMessage responseMessage = this.buildResponseMessage(e, message);

            // after intercept
            if (this.interceptors != null && !interceptors.isEmpty()) {
                Iterator<RpcHandleInterceptor> iterator = this.interceptors.iterator();
                while (iterator.hasNext()) {
                    iterator.next().after(responseMessage);
                }
            }

            try {
                writeChannel(responseMessage, e);
            } catch (Exception ex) {
                log.error("Write net channel error", ex);
            }
        } finally {
            // error intercept
            if (this.interceptors != null && !interceptors.isEmpty()) {
                Iterator<RpcHandleInterceptor> iterator = this.interceptors.iterator();
                while (iterator.hasNext()) {
                    iterator.next().doLast(e);
                }
            }
        }
    }

    /**
     * @return
     * @Author 1137
     * @Description 处理异常
     * @Date
     * @Param
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
        NetResponseMessage responseMessage = this.exceptionHandler.handle(e);
        try {
            this.writeChannel(responseMessage, e);
        } catch (RelayJobNetException e1) {
            log.error("Network response channel write error", e1);
        }
    }

    private void writeChannel(NetResponseMessage responseMessage, ChannelEvent e) throws RelayJobNetException {
        // response
        try {
            HttpResponse httpResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.valueOf(responseMessage.getStatus())) {
                @Override
                public HttpHeaders headers() {
                    DefaultHttpHeaders headers = new DefaultHttpHeaders();
                    Map<String, String> messageHeaders = responseMessage.getHeaders();
                    if (messageHeaders != null && !messageHeaders.isEmpty()) {
                        messageHeaders.entrySet().stream().forEach(entry -> headers.add(entry.getKey(), entry.getValue()));
                    }
                    return headers;
                }
            };
            ChannelBuffer responseBuffer = new DynamicChannelBuffer(256);
            responseBuffer.writeBytes(this.messageConvert.writeConvert(responseMessage.getMessage()));
            httpResponse.setContent(responseBuffer);

            Channel channel = e.getChannel();

            // response client
            ChannelFuture future = channel.write(httpResponse);

            // add channel listener, close channel
            future.addListener(f -> f.getChannel().close());
        } finally {
            try {
                e.getChannel().close();
            } catch (Exception e1) {
                log.error("Close net channel error.", e1);
            }
        }
    }


    private NetRequestMessage buildRequestMessage(MessageEvent e) throws RelayJobNetException {
        HttpRequest request = (HttpRequest) e.getMessage();
        String uri = request.getUri();

        // 读取报文
        byte[] array = request.getContent().array();

        HttpHeaders headers = request.headers();
        Set<String> names = headers.names();
        Map<String, String> requestHeader = new TreeMap<>();
        if (names != null && !names.isEmpty()) {
            names.stream().forEach(head -> requestHeader.put(head, headers.get(head)));
        }

        SocketAddress socketAddress = e.getRemoteAddress();

        InetSocketAddress address = null;
        if (socketAddress instanceof InetSocketAddress) {
            address = (InetSocketAddress) socketAddress;
        }

        return NetRequestMessage.builder()
                .uri(uri)
                .message(this.messageConvert.readConvert(array))
                .headers(requestHeader)
                .socketAddress(address)
                .build();
    }

    private NetResponseMessage buildResponseMessage(MessageEvent e, byte[] message) {
        return NetResponseMessage.builder()
                .headers(new TreeMap<>())
                .message(message)
                .status(HttpResponseStatus.OK.getCode())
                .build();
    }

}
